using Confluent.Kafka;
using CoreManager.MessageQueue.Models;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;

namespace CoreManager.MessageQueue.Services;

/// <summary>
/// Kafka生产者服务实现
/// </summary>
public class KafkaProducerService : IKafkaProducerService, IDisposable
{
    private readonly IProducer<string, string> _producer;
    private readonly ILogger<KafkaProducerService> _logger;
    private readonly KafkaOptions _options;
    private bool _disposed;

    public KafkaProducerService(IOptions<KafkaOptions> options, ILogger<KafkaProducerService> logger)
    {
        _options = options.Value;
        _logger = logger;

        var config = new ProducerConfig
        {
            BootstrapServers = _options.BootstrapServers,
            Acks = Enum.Parse<Acks>(_options.Producer.Acks, true),
            MessageSendMaxRetries = _options.Producer.Retries,
            BatchSize = _options.Producer.BatchSize,
            CompressionType = Enum.Parse<CompressionType>(_options.Producer.CompressionType, true),
            EnableIdempotence = true,
            MaxInFlight = 1,
            MessageTimeoutMs = 30000
        };

        _producer = new ProducerBuilder<string, string>(config)
            .SetErrorHandler((_, error) =>
            {
                _logger.LogError("Kafka producer error: {Error}", error.Reason);
            })
            .SetLogHandler((_, log) =>
            {
                _logger.LogInformation("Kafka producer log: {Message}", log.Message);
            })
            .Build();

        _logger.LogInformation("Kafka producer initialized with servers: {Servers}", _options.BootstrapServers);
    }

    /// <inheritdoc />
    public async Task<MessageSendResult> SendMessageAsync(string topic, KafkaMessage message, string? key = null, CancellationToken cancellationToken = default)
    {
        ThrowIfDisposed();

        try
        {
            var jsonMessage = JsonConvert.SerializeObject(message, Formatting.None);
            var kafkaMessage = new Message<string, string>
            {
                Key = key ?? message.MessageId,
                Value = jsonMessage,
                Timestamp = new Timestamp(message.Timestamp)
            };

            var deliveryResult = await _producer.ProduceAsync(topic, kafkaMessage, cancellationToken);

            _logger.LogInformation("Message sent successfully to topic {Topic}, partition {Partition}, offset {Offset}", 
                topic, deliveryResult.Partition.Value, deliveryResult.Offset.Value);

            return new MessageSendResult
            {
                IsSuccess = true,
                Topic = topic,
                Partition = deliveryResult.Partition.Value,
                Offset = deliveryResult.Offset.Value,
                MessageId = message.MessageId
            };
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to send message to topic {Topic}", topic);
            return new MessageSendResult
            {
                IsSuccess = false,
                Topic = topic,
                ErrorMessage = ex.Message,
                MessageId = message.MessageId
            };
        }
    }

    /// <inheritdoc />
    public async Task<MessageSendResult> SendMessageAsync<T>(string topic, T data, string? key = null, CancellationToken cancellationToken = default)
    {
        var message = new KafkaMessage<T>
        {
            MessageType = typeof(T).Name,
            Data = data
        };

        return await SendMessageAsync(topic, message, key, cancellationToken);
    }

    /// <inheritdoc />
    public async Task<List<MessageSendResult>> SendMessagesAsync(string topic, IEnumerable<KafkaMessage> messages, CancellationToken cancellationToken = default)
    {
        var results = new List<MessageSendResult>();
        var tasks = messages.Select(message => SendMessageAsync(topic, message, null, cancellationToken));
        
        var sendResults = await Task.WhenAll(tasks);
        results.AddRange(sendResults);

        return results;
    }

    /// <inheritdoc />
    public void Dispose()
    {
        if (_disposed) return;

        try
        {
            _producer?.Flush(TimeSpan.FromSeconds(10));
            _producer?.Dispose();
            _logger.LogInformation("Kafka producer disposed");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error disposing Kafka producer");
        }
        finally
        {
            _disposed = true;
        }
    }

    private void ThrowIfDisposed()
    {
        if (_disposed)
        {
            throw new ObjectDisposedException(nameof(KafkaProducerService));
        }
    }
}
